package bigdata.hdfs;
//Copy all files from a folder on the local computer to the HDFS system
//请编写代码，将本地计算机上某个文件夹下的所有文件复制到hdfs系统

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;

import org.apache.hadoop.fs.*;

public class myPut {
	
	String localpath=null;
	String remotepath=null;

	public myPut(String localpath,String remotepath) {
		this.localpath=localpath;
		this.remotepath=remotepath;
	}
	
	public File[] getFiles() {
		File directory=new File(localpath);
		if(directory.exists() && directory.isDirectory()) {
			File[] files=directory.listFiles();
			return files;
		}
		else 
			return null;
	}
	
	public void doPut() throws FileNotFoundException, IOException {
		File[] files=getFiles();
		Connect2HDFS.connect2hdfs();
		FileSystem fs=FileSystem.get(Connect2HDFS.getconf());
		FSDataOutputStream out=null;
		if(files!=null) {
			for (File file : files) {
                if (file.isFile() && file.getName().endsWith("txt")) { // 确保是文件而非目录
                	String hdfsfilename = this.remotepath + "/" + file.getName();
                	out=fs.create(new Path(hdfsfilename));
                	
                	System.out.println("file name:"+ file.getPath());
                	try (BufferedReader br = new BufferedReader(new FileReader(file.getPath()))) {
						String line;
						while ((line=br.readLine())!=null) {
							out.writeBytes(line);
							//System.out.println(line);
						}
						//System.out.println();
						out.close();
					}
                }
			}
		}  
	}
	
	public static void main(String[] args) throws IOException{
		// TODO Auto-generated method stub
		
		String localpath="/home/abc";
		String remotepath="hdfspath";
		
		myPut put=new myPut(localpath,remotepath);
		put.doPut();
	}
}